package com.zs.toolbox.web2j.system.message.mq;

import com.alibaba.fastjson.JSON;
import com.zs.toolbox.web2j.system.message.domain.request.WtSendEmailRequest;
import com.zs.toolbox.web2j.system.message.domain.request.WxSendSmsVcodeRequest;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.starter.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

import static com.zs.toolbox.web2j.common.exception.WtExceptions.E_MESSAGE_SEND_ERROR;

/**
 * @author shuai.zhang  on 2018/9/26
 * @description
 */
@Component
@Slf4j
public class WtRocketMQMessageProductImpl implements WtMessageProduct {


    @Value("${message.queue.rocketmq.nameserver}")
    String nameServer;

    @Value("${message.queue.rocketmq.group}")
    String productGroup;

    @Value("${message.queue.rocketmq.topic}")
    String productTopic;

    Long sendTimeout = 3000L;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

//    DefaultMQProducer producer;

  /*  @PostConstruct
    public void init() throws MQClientException {
        producer = new DefaultMQProducer(productGroup);
        producer.setNamesrvAddr(nameServer);
        producer.start();
        log.info("message product rocketmq nameserver<{}>,group<{}> start finish [ok]", nameServer, productGroup);
    }

    @PreDestroy
    public void deinit() {
        producer.shutdown();
        log.info("message product rocketmq nameserver<{}>,group<{}> shutdown finish [ok]", nameServer, productGroup);
    }*/

    @Override
    public boolean sendEmailAsync(WtSendEmailRequest request) {
        request.setIsSync(true);
        return send(request, WtMessageTagConst.MESSSAGE_TAG_EMAIL, sendTimeout);
    }

    @Override
    public boolean sendSmsRegisterVcodeAsync(WxSendSmsVcodeRequest request) {
        request.setIsSync(true);
        return send(request, WtMessageTagConst.MESSSAGE_TAG_SMS_REGISTER, sendTimeout);
    }

    @Override
    public boolean sendSmsLoginVcodeAsync(WxSendSmsVcodeRequest request) {
        request.setIsSync(true);
        return send(request, WtMessageTagConst.MESSSAGE_TAG_SMS_LOGIN, sendTimeout);
    }

    @Override
    public boolean sendSmsFindPasswordVcodeAsync(WxSendSmsVcodeRequest request) {
        request.setIsSync(true);
        return send(request, WtMessageTagConst.MESSSAGE_TAG_SMS_FINDPASSWORD, sendTimeout);
    }

    public boolean send(Object body, String tag, long timeout) {
        try {
            String destnation = String.join(":", productTopic, tag);

            rocketMQTemplate.asyncSend(destnation, JSON.toJSONString(body), new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.debug("send message to queue success topic: {} result: {}", productTopic, sendResult);
                }

                @Override
                public void onException(Throwable e) {
                    log.warn("send message to queue success error topic: {},message {}", productTopic, e.getMessage());
                    log.debug(String.format("send message error topic %s", productGroup), e);
                }
            },timeout);
            return true;
        } catch (Exception e) {
            log.error("message rocketmq nameserver<{}>,group<{}> error message:{}", nameServer, productGroup, e.getMessage());
            log.debug("message rocket send message error", e);
            throw E_MESSAGE_SEND_ERROR;
        }
    }
}
